package com.csw.flink.window

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object Demo04WindowProcess {
  def main(args: Array[String]): Unit = {

    /*
    //如果用户001按以下顺序进行操作则发出警告
    001,a
    001,b
    001,c
    001,d
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //创建配置文件对象
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    /**
      * 创建kafka消费者
      */
    val consumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("test_topicl", new SimpleStringSchema(), properties)

    //读取最新数据
    consumer.setStartFromLatest()

    val kafkaDS: DataStream[String] = env.addSource(consumer)

    val kvDS: KeyedStream[(String, String), String] = kafkaDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1))
    }).keyBy(_._1)

    //如果5秒没有数据则进行计算
    val windowDS: WindowedStream[(String, String), String, TimeWindow] = kvDS.window(
      ProcessingTimeSessionWindows.withGap(Time.seconds(5)))

    //使用flink底层api操作底层事件、时间和状态
    val countDS: DataStream[(String, String)] = windowDS.process(new MyWindowProcessFunction)

    countDS.print()

    env.execute()
  }
}

class MyWindowProcessFunction extends ProcessWindowFunction[(String, String), (String, String), String, TimeWindow] {
  /**
    * 处理窗口内数据的方法，每一个key在每一个窗口内处理一次
    *
    * @param key      数据的key
    * @param context  上下文对象
    * @param elements 一个key在一个窗口内的所有数据
    * @param out      用于将数据发送到下游
    */
  override def process(key: String, context: Context, elements: Iterable[(String, String)], out: Collector[(String, String)]): Unit = {
    //如果一个人操作的顺序为a,b,c,d则发出警告

    val flag = List("a", "b", "c", "d")

   val list: List[String] = elements.toList.map(_._2)

    var boolean = true

    if (list.length >= 4) {

      var i = 0
      while (i < flag.length) {

        val f = flag(i)
        val l = list(i)

        if (!f.equals(l)) {
          boolean = false
        }
        i += 1
      }
    }

    if (boolean){
      //发出警告
      out.collect(key,list.mkString("|"))
    }
  }
}
